package com.etc.sparkstrea

import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf}

/**
  * 基于updateStateByKey算子实现缓存机制的实时wordcount程序
  * @author Administrator
  *
  */
object StreamingUpdateByKey {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("StreamingUpdateByKey").setMaster("local[*]")
    val sc = new StreamingContext(conf, Seconds(5))
    sc.checkpoint("hdfs://master:9000/wordcount_checkpoint")
    val line = sc.socketTextStream("192.168.88.5",8888)
    line.flatMap(_.split(" ")).map(a => (a, 1))
      //state: Option[Int]判断以前是否出现过出现过就累加
      .updateStateByKey((values: Seq[Int], state: Option[Int]) => {
      //没出现过就初始化为0
      var newValues = state.getOrElse(0)
      //循环累加
      for (value <- values) {
        newValues += value
      }
      Option(newValues)
    }).print()
    sc.start()
    sc.awaitTermination()
  }
}
